package com.langshixiaobai.mqdemo.server.datacenter;

import com.langshixiaobai.mqdemo.common.BinaryTool;
import com.langshixiaobai.mqdemo.common.MqException;
import com.langshixiaobai.mqdemo.server.core.MSGQueue;
import com.langshixiaobai.mqdemo.server.core.Message;

import java.io.*;
import java.util.Date;
import java.util.LinkedList;
import java.util.List;
import java.util.Scanner;

public class MessageFileManager {

    //进行初始化
    public void init() {
        //先创建放在这，方便后续可以拓展
    }

    //统计数据信息类
    public static class Stat {
        //当前队列消息总数
        public int totalCount;
        //当前队列有效消息总数
        public int validCount;
    }


    //获得队列对应文件所在的目录
    public String getFileDic(String queueName) {
        return "./data/" + queueName;
    }

    //获得队列对应数据文件所在的路径
    public String getFileDataPath(String queueName) {
        //一般二进制文件不用txt后缀，txt表示文本文件，这里简化了
        return getFileDic(queueName) + "/queue_data.txt";
    }

    //获得队列对应统计数据信息的文件所在的路径
    public String getFileStatPath(String queueName) {
        return getFileDic(queueName) + "/queue_stat.txt";
    }

    //从硬盘中读取文件
    public Stat readStat(String queueName) {
        Stat stat = new Stat();
        //获取当前队列统计消息文件
        String fileStatPath = getFileStatPath(queueName);
        //InputStream读取文件，会自动关闭
        try (FileInputStream fileInputStream = new FileInputStream(fileStatPath)) {
            Scanner scanner = new Scanner(fileInputStream);
            //获取当前队列消息总数
            stat.totalCount = scanner.nextInt();
            //获取当前队列有效消息总数
            stat.validCount = scanner.nextInt();
            return stat;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    //向硬盘中写入统计文件
    public void writeStat(String queueName, Stat stat) {
        //获取当前队列文件路径
        String fileStatPath = getFileStatPath(queueName);
        //append 属性默认为false，每当打开文件就会将内容覆盖，重新写入
        try (FileOutputStream fileOutputStream = new FileOutputStream(fileStatPath)) {
            PrintWriter printWriter = new PrintWriter(fileOutputStream);
            //写入文件 \t进行分割
            String data = stat.totalCount + "\t" + stat.validCount;
            //此时数据被写到缓存中
            printWriter.write(data);
            //将缓存区中数据内容刷新到硬盘中
            printWriter.flush();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    //初始化，将目录，队列消息文件，消息统计信息文件都创建出来
    public void createQueueFile(String queueName) throws IOException {
        String dic = getFileDic(queueName);
        File file = new File(dic);
        if (!file.exists()) {
            boolean ok = file.mkdirs();
            if (!ok) {
                throw new IOException("创建目录失败:" + queueName);
            }
        }
        String fileDataPath = getFileDataPath(queueName);
        file = new File(fileDataPath);
        if (!file.exists()) {
            boolean ok = file.createNewFile();
            if (!ok) {
                throw new IOException("创建队列消息文件错误：" + queueName);
            }
        }

        //创建统计信息文件
        Stat stat = new Stat();
        stat.totalCount = 0;
        stat.validCount = 0;
        //将stat文件初始化，默认都为0
        writeStat(queueName, stat);
    }

    //销毁文件
    public void destroyQueueFiles(String queueName) throws IOException {
        String fileStatPath = getFileStatPath(queueName);
        String fileDataPath = getFileDataPath(queueName);
        String dic = getFileDic(queueName);
        File file = new File(fileStatPath);
        boolean ok1 = file.delete();
        file = new File(fileDataPath);
        boolean ok2 = file.delete();
        file = new File(dic);
        boolean ok3 = file.delete();
        if (!ok1 || !ok2 || !ok3) {
            throw new IOException("删除失败:" + file.getAbsolutePath());
        }
    }

    public boolean checkQueueFiles(String queueName) {
        String fileStatPath = getFileStatPath(queueName);
        String fileDataPath = getFileDataPath(queueName);
        File fileDataFile = new File(fileDataPath);
        File fileStatfFile = new File(fileStatPath);
        if (!fileStatfFile.exists()) {
            return false;
        }
        if (!fileDataFile.exists()) {
            return false;
        }
        return true;
    }

    //存入消息
    public void sendMessage(MSGQueue queue, Message message) throws MqException {
        boolean ok = checkQueueFiles(queue.getName());
        if (!ok) {
            throw new MqException("【MessageFileManager】文件不存在");
        }
        //将其序列化，为后续写入文件作准备
        byte[] bytes = BinaryTool.toBytes(message);
        //上锁防止线程安全问题
        synchronized (queue) {
            File file = new File(getFileDataPath(queue.getName()));
            //OffsetBegin = 长度+4（用于记录消息长度）
            //容易发生线程安全问题
            message.setBeginOffset(file.length() + 4);
            message.setEndOffset(file.length() + 4 + bytes.length);

            try (OutputStream objectOutputStream = new FileOutputStream(file, true)) {
                //如果直接写入，即使是int 也只占用1个字节
                // objectOutputStream.write(bytes.length);
                //解决了上述问题，将int写入时采用4个字节
                try (DataOutputStream dataOutputStream = new DataOutputStream(objectOutputStream)) {
                    //将一个字节变量使用四个字节写入
                    dataOutputStream.writeInt(bytes.length);
                    //一个字节一个字节写入
                    dataOutputStream.write(bytes);
                }
            } catch (IOException e) {
                throw new RuntimeException(e);
            }

            Stat stat = readStat(queue.getName());
            stat.totalCount += 1;
            stat.validCount += 1;
            writeStat(queue.getName(), stat);
        }
    }

    //随机写入
    public void deleteMessageFile(MSGQueue queue, Message message) {
        String fileDataPath = getFileDataPath(queue.getName());
        try (RandomAccessFile randomAccessFile = new RandomAccessFile(fileDataPath, "rw")) {
            byte[] bufferSrc = new byte[(int) (message.getEndOffset() - message.getBeginOffset())];
            //移动光标到开始闻之
            randomAccessFile.seek(message.getBeginOffset());
            //读取
            //读入到字节数组中，读的时候光标也会移动
            randomAccessFile.read(bufferSrc);
            //转换成Message消息对象
            Message queueMessage = (Message) BinaryTool.formBytes(bufferSrc);
            //将消息进行逻辑删除，设置为失效
            queueMessage.setIsValid((byte) 0);
            //将修改后的消息写入
            //移动光标到开始位置
            randomAccessFile.seek(message.getBeginOffset());
            //写入
            randomAccessFile.write(BinaryTool.toBytes(queueMessage));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        Stat stat = readStat(queue.getName());
        if (stat.validCount > 0) {
            stat.validCount -= 1;
        }
        writeStat(queue.getName(), stat);

    }

    //LinkedList 方便后续在内存中拿去/删除节点
    //loadAllMessageFromQueue读取文件中有效消息
    public List<Message> loadAllMessageFromQueue(String queueName) {
        List<Message> list = new LinkedList<>();
        long beginOffset = 0;
        try (InputStream fileInputStream = new FileInputStream(getFileDataPath(queueName))) {
            //按字节读取
            try (DataInputStream dataInputStream = new DataInputStream(fileInputStream)) {
                while (true) {
                    //一次读取4个字节，正好消息长度对应也为4个字节
                    //当文件数据都被读取完成时，会抛出EOFException异常，通过捕获判断是否读取完毕
                    int messageSize = dataInputStream.readInt();
                    //根据读出的长度初始化数组
                    byte[] bytes = new byte[messageSize];
                    //实际读到的长度
                    int actualLength = dataInputStream.read(bytes);
                    if (messageSize != actualLength) {
                        throw new MqException("【MessageFIleManager】文件格式错误");
                    }

                    Message message = (Message) BinaryTool.formBytes(bytes);
                    if (message.getIsValid() != 0x1) {
                        beginOffset += (4 + messageSize);
                        continue;
                    }
                    //设置偏移量
                    message.setBeginOffset(beginOffset + 4);
                    message.setEndOffset(beginOffset + 4 + messageSize);
                    //计算偏移量
                    beginOffset += 4 + messageSize;
                    //存储
                    list.add(message);
                }

            }
        } catch (EOFException e) {
            System.out.println("[MessageFileManager]文件读取完成");
            return list;
        } catch (IOException | MqException e) {
            throw new RuntimeException(e);
        }
    }

    //复制删除
    //判断是否要进行GC垃圾回收
    public boolean checkGC(String queueName) {
        //先获取消息统计信息中，消息总数和有效消息
        Stat stat = readStat(queueName);
        if (stat.totalCount > 2000 && (double) stat.validCount / (double) stat.totalCount < 0.5) {
            return true;
        }
        return false;
    }

    //删除原理，通过创建新的文件，将有效消息复制到新文件，然后删除旧文件，将新文件代替旧文件
    public String getNewFilePath(String queueName) {
        return getFileDic(queueName) + "/queue_data_new.txt";
    }

    //GC回收机制
    //删除原理，通过创建新的文件，将有效消息复制到新文件，然后删除旧文件，将新文件代替旧文件
    public void gc(MSGQueue queue) throws MqException, IOException {
        //上锁，与前面写入消息所有锁相同，当gc时需要禁止写入消息，防止顺序错乱
        synchronized (queue) {
            //开始gc时间
            long begin = System.currentTimeMillis();
            //获取 new File新文件路径
            File newFile = new File(getNewFilePath(queue.getName()));
            //如果已经存在新文件说明上次GC回收出现问题，可能中途中断
            if (newFile.exists()) {
                throw new MqException("【MessageFileManager】gc时发现queue_data_new.txt文件已经存在: " + queue.getName());
            }
            //创建新文件是否成功
            boolean ok = newFile.createNewFile();
            //不成功要抛出异常
            if (!ok) {
                throw new MqException("【MessageFileManager】gc时 创建queue_data_new.txt文件时，创建失败: " + queue.getName());
            }
            //获取有效消息
            List<Message> messages = loadAllMessageFromQueue(queue.getName());
            //将有效消息写入新文件
            try (OutputStream outputStream = new FileOutputStream(newFile)) {
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    for (Message message : messages) {
                        //写入新文件
                        byte[] bytes = BinaryTool.toBytes(message);
                        dataOutputStream.writeInt(bytes.length);
                        dataOutputStream.write(bytes);
                    }
                }
            }
            //替换消息数据文件
            //删除旧文件
            File oldFile = new File(getFileDataPath(queue.getName()));
            ok = oldFile.delete();
            if (!ok) {
                throw new MqException("【MessageFileManager】删除oldFile错误");
            }
            //将新文件名改成为旧文件名，替代旧文件成为新的消息文件queue_data.txt
            ok = newFile.renameTo(oldFile);
            if (!ok) {
                throw new MqException("【MessageFileManager】替换消息文件失败");
            }
            //更新消息统计文件
            //先从文件读取
            Stat stat = readStat(queue.getName());
            //重新统计消息数量
            stat.validCount = messages.size();
            stat.totalCount = messages.size();
            //重新写入
            writeStat(queue.getName(), stat);

            //gc结束时间
            long end = System.currentTimeMillis();
            //通过记录gc所需时间，来判断当前gc是否耗时，便于后续优化
            System.out.println("【MessageFileManager】中队列：" + queue.getName() + " gc回收所消耗时间为：" + (end - begin));
        }
    }
}
